package com.kkb.sparkdemo

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SparkSession}


//case class Exam(class_id: Int, student_name: String, age: Int, sex: String, subject: String, score: Int)

object SparkExerciseSQL {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("sparkDSL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //加载数据
    val rdd1 = sc.textFile(this.getClass.getClassLoader.getResource("test.txt").getPath).map(x => x.split(" "))

    //把rdd与样例类进行关联
    val examRDD = rdd1.map(x => Person(x(0), x(1), x(2).toInt, x(3), x(4), x(5).toInt)
    //把rdd转换成DataFrame
    import spark.implicits._ // 隐式转换

    val examDF = examRDD.toDF
    //打印schema信息
    examDF.printSchema
    //展示数据
    examDF.show
    
    //DataFrame注册成表
    examDF.createTempView("exam")

    //使用SparkSession调用sql方法统计查询

    //一共有多少个小于20岁的人参加考试
    spark.sql("select count(*) from exam where age < 20").show

    //一共有多少个等于20岁的人参加考试
    spark.sql("select count(*) from exam where age = 20").show

    //一共有多少个大于20岁的人参加考试
    spark.sql("select count(*) from exam where age > 20").show

    //一共有多个男生参加考试？
    spark.sql("select count(*) from exam where sex = '男'").show

    //一共有多少个女生参加考试？
    spark.sql("select count(*) from exam where sex = '女'").show

    //12班有多少人参加考试？
    spark.sql("select count(*) from exam where class_id = 12").show

    //13班有多少人参加考试？
    spark.sql("select count(*) from exam where class_id = 13").show

    //语文科目的平均成绩是多少？
    spark.sql("select avg(score) from exam where subject = 'chinese'").show

    //数学科目的平均成绩是多少？
    spark.sql("select avg(score) from exam where subject = 'math'").show

    //英语科目的平均成绩是多少？
    spark.sql("select avg(score) from exam where subject = 'english'").show

    //每个人平均成绩是多少？
    spark.sql("select avg(score) from exam ").show

    //12班平均成绩是多少？
    spark.sql("select avg(score) from exam where class_id = 12 ").show

    //12班男生平均总成绩是多少？
    spark.sql("select avg(score) from exam where class_id = 12 and sex = '男' ").show

    //12班女生平均总成绩是多少？
    spark.sql("select avg(score) from exam where class_id = 12 and sex = '女' ").show

    //13班平均成绩是多少？
    spark.sql("select avg(score) from exam where class_id = 13 ").show

    //13班男生平均总成绩是多少？
    spark.sql("select avg(score) from exam where class_id = 13 and sex = '男' ").show

    //13班女生平均总成绩是多少？
    spark.sql("select avg(score) from exam where class_id = 13 and sex = '女' ").show

    //全校语文成绩最高分是多少？
    spark.sql("select max(score) from exam where subject = 'chinese'").show

    //12班语文成绩最低分是多少？
    spark.sql("select min(score) from exam where class_id = 12 and subject = 'chinese'").show

    //13班数学最高成绩是多少？
    spark.sql("select max(score) from exam where class_id = 13 and subject = 'math'").show

    //总成绩大于150分的12班的女生有几个？
    spark.sql("select COUNT(*) FROM (select count(*) from exam where class_id = 12 and sex = '女' group by student_name having sum(score)>150) ex").show

    //总成绩大于150分，且数学大于等于70，且年龄大于等于19岁的学生的平均成绩是多少？
    spark.sql("SELECT student_name, AVG(score) FROM exam WHERE student_name IN(SELECT student_name FROM exam WHERE age>=19 and student_name NOT IN(SELECT student_name FROM exam WHERE SUBJECT = 'math' AND score < 70) GROUP BY class_id, student_name HAVING SUM(score)> 150) GROUP BY student_name\n\n").show

    spark.stop()
  }
} 